package com.sheepone.distributor.thread;

import java.util.concurrent.*;

/**
 * @author zhonghuashishan
 */
public class SafeThreadPool {

    private final Semaphore semaphore;

    private final ThreadPoolExecutor threadPoolExecutor;

    public SafeThreadPool(String name, int permits, int keepAliveTime) {
        semaphore = new Semaphore(permits);
        threadPoolExecutor = new ThreadPoolExecutor(
                permits,
                permits * 2,
                keepAliveTime,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(), // 可能会出现重复消费
                NamedDaemonThreadFactory.getInstance(name)
        );
    }

    public int getActiveCount(){
        return threadPoolExecutor.getActiveCount();
    }

    public void submit(Runnable task) {
        semaphore.acquireUninterruptibly();
        threadPoolExecutor.submit(() -> {
            try {
                task.run();
            } finally {
                semaphore.release();
                // 不安全现象：
                // 可能存在 permits 个线程在这里同时释放 semaphore，同时另外还有 >= permits 个线程 在 acquire 中
                // 这将导致 原来持有 semaphore 的线程还未结束，而 等待的线程获取到 semaphore 后 涌入（最多涌入 permits 个）
                // 导致 ThreadPool 满载，新来的线程被拒绝从而抛出异常
            }
        });
    }

    public void shotDown() {
        threadPoolExecutor.shutdown();;
    }
}